package com.jida.hadoop.mr.TopN;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.jida.hadoop.mr.tools.TProperties;


/** 
 * @����   mr_job.dx 
 * @�ļ��� DxTopN.java 
 * @�汾 V 1.0 
 * @�������� �û���Ϊȡ����topN
 */
public class DxTopN {

	/**
	 * @param 
	 * args[0]������·�������˵�ַͳ�ƽ�����ݣ�
	 * args[1]��������·��
	 * args[2]�������ֶΣ��û�id��=1
	 * args[3]������ֵ�����ʴ�����=2
	 * args[4]��topȡֵ���Χ   =5
	 * Description��mapreduceͳ���û��������ŵ�ַ��ȡǰN����¼
	 * */
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		//����map��reduce�ڴ�ʹ�ô�С
		conf.set("mapreduce.map.memory.mb", "3072");
		conf.set("mapreduce.reduce.memory.mb", "2048");
		//��������
		//�����ֶΣ��û�ID
		conf.set("topkey", "1");
		//����ֵ��PV������
		conf.set("topvalue", "2");
		//����ȡǰN����¼��ǰ����
		conf.set("topn", "5");
		Job job =Job.getInstance(conf, "Dx_TopN");
		job.setJarByClass(DxTopN.class);
		//map��reduce����
		job.setMapperClass(TopNMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setReducerClass(TopNReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		//����·��
		FileInputFormat.addInputPath(job, new Path("file:///D:/测试数据/dx/03.DxPathStatistic/part-r-*"));
		//���·��
		FileOutputFormat.setOutputPath(job, new Path("file:///D:/测试数据/dx/10.DxTopN"));
		//�ύ��ҵ �ж��˳�������0�����˳���1�������˳���
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

	/**
	 * ��ָ���ֶ���Ϊkey����
	 * */
	public static class TopNMapper extends Mapper<LongWritable, Text, Text, Text> {
		private String topkey ;
		private Text okey = new Text();
		//��ȡ�����ֶ�λ��
		@Override
		protected void setup( Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
			topkey = context.getConfiguration().get("topkey");
		}

		//�Է����ֶ�Ϊkey
		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] values = value.toString().split(TProperties.getValue("fileoutsplit"));
			okey = new Text(values[Integer.parseInt(topkey)]);
			//���û�����
			context.write(okey, value);
		}
	}

	/**
	 * ����ȡtopN����¼
	 * */
	public static class TopNReducer extends Reducer<Text, Text, NullWritable, Text> {
		private Text result = new Text();
		private int topvalue ;
		private int topn ;
		//��ȡ����
		@Override
		protected void setup( Reducer<Text, Text, NullWritable, Text>.Context context) throws IOException, InterruptedException {
			//����ֵ�����ʴ���
			topvalue = Integer.parseInt(context.getConfiguration().get("topvalue"));
			//ȡNֵ��Ĭ��ֵΪ5
			topn = Integer.parseInt(context.getConfiguration().get("topn"));
		}

		//reduce����ȡֵ
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
			TreeMap<Integer, String> topMap = new TreeMap<Integer, String>();
			//ѭ������
		    for (Text val : values) {
			String[] str = val.toString().split(TProperties.getValue("fileoutsplit"));
			//�����ݷ������ݼ���
			topMap.put(Integer.parseInt(str[topvalue]), val.toString());
			//�Ƴ�����ȡֵ�����Ľ�С����
			if(topMap.size() > topn){
				topMap.remove(topMap.firstKey());
			}
		    }
		    int num = 0;
		    Iterator<Integer> iterator = topMap.keySet().iterator();
		    //����ת������������
		    List<String> list = new ArrayList<String>();
		    while (iterator.hasNext()) {
			list.add(topMap.get(iterator.next()));
		    }
		    //ȡ��������+1�����ڼ�������
		    num = list.size() + 1 ;
		    for(int i = 0 ;i< list.size() ; i++){
			//����
			num = num -1 ;
			//��ΪID���û�ID��PV�����ʴ�����������
			result = new Text(list.get(i) + TProperties.getValue("outfilesplit") + num );
				context.write(NullWritable.get(), result);
		    }
		}
	}
}
